from pyflink.common import Encoder, Types, SimpleStringSchema, WatermarkStrategy, Time
from pyflink.datastream import StreamExecutionEnvironment, RuntimeExecutionMode
from pyflink.datastream.connectors.file_system import FileSink, RollingPolicy
from pyflink.datastream.connectors.kafka import KafkaOffsetsInitializer, KafkaSource
from pyflink.datastream.window import SlidingProcessingTimeWindows, TumblingProcessingTimeWindows

# 1、创建flink执行环境
env = StreamExecutionEnvironment.get_execution_environment()

env.set_parallelism(1)

source = KafkaSource.builder() \
    .set_bootstrap_servers("master:9092") \
    .set_topics("window") \
    .set_group_id("my-group") \
    .set_starting_offsets(KafkaOffsetsInitializer.latest()) \
    .set_value_only_deserializer(SimpleStringSchema()) \
    .build()

# 读取数据
lines_ds = env.from_source(source, WatermarkStrategy.no_watermarks(), "Kafka Source")

kv_ds = lines_ds.flat_map(lambda line: line.split(",")).map(lambda word: (word, 1))

window_ds = kv_ds.window_all(TumblingProcessingTimeWindows.of(Time.seconds(5)))

count_ds = window_ds.reduce(lambda value1, value2: (value1[0], value1[1] + value2[1]))
count_ds.print()

env.execute()